package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {


  def main(args: Array[String]): Unit = {

    //1、创建spark 配置文件对象
    val conf = new SparkConf()

    //local  本地测试模式   提交到集群运行需要删掉
    conf.setMaster("local")

    conf.setAppName("Demo1WordCount") //job名  可以在yarn ui里面看见


    //2、创建spark上下文对象 入口
    val sc = new SparkContext(conf)


    //3、读取文件

    /**
      * textFile
      *
      * 如果是本地运行  会读取本地文件
      * 如果提交到集群运行 会读取hdfs文件
      *
      */
    //RDD   弹性的分布式数据集   现在可以理解为一个集合
    //    val linesRDD: RDD[String] = sc.textFile("spark/data/words.txt")
    val linesRDD: RDD[String] = sc.textFile("spark/data/words.txt")


    //4、将单词拆分出来
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    wordsRDD.saveAsTextFile("spark/data/out1")

     //5、转换成kv格式
     val kvRDD = wordsRDD.map(word => (word, 1))

     //通过key对value进行聚合  需要一个聚合函数
     //(x, y) => x + y)   聚合的方式
     val countRDD = kvRDD.reduceByKey((x, y) => x + y)

     //countRDD.foreach(println)


     //将结果保存到hdfs
     countRDD.saveAsTextFile("spark/data/out")
  }
}
